package dws

import java.nio.charset.Charset
import java.util

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.{Cell, HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Result, ResultScanner, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes

class OrderHBaseReader extends RichSourceFunction[(String, String)]{
  var connection: Connection = _
  var hTable: Table = _
  var scan: Scan = _
  val tableName: String = "lagou_trade_orders"
  val family: String = "f1"


  override def open(parameters: Configuration): Unit = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort","2181")
    connection = ConnectionFactory.createConnection(configuration)
    hTable = connection.getTable(TableName.valueOf(tableName))
    scan = new Scan()
    scan.addFamily(family.getBytes(Charset.forName("UTF-8")))
  }

  override def run(ctx: SourceFunction.SourceContext[(String, String)]): Unit = {
    val result: ResultScanner = hTable.getScanner(scan)
    val iter: util.Iterator[Result] = result.iterator()
    while (iter.hasNext) {
      val record: Result = iter.next()
      val rowKey: String = Bytes.toString(record.getRow)
      val cells: util.List[Cell] = record.listCells()
      val stringBuilder = new StringBuilder
      cells.forEach(cell => {
        stringBuilder.append(Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append(",")
      })
      stringBuilder.deleteCharAt(stringBuilder.length - 1)
      ctx.collect((rowKey, stringBuilder.toString()))
    }
  }

  override def cancel(): Unit = {
    try {
      if(hTable != null) {
        hTable.close()
      }
      if(connection != null) {
        connection.close()
      }
    } catch {
      case e:Exception => println(e.getMessage)
    }
  }
}
